package io.renren.socket.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.renren.htby.utils.SocketUtils;

/**
 * 分发系统的配置 NIOServer
 * 
 * @author lfy.xys
 * @date 2018年5月30日
 *
 */
public class DisConfigNIOServer implements Runnable {
	private int port;
	private ByteBuffer rBuffer = ByteBuffer.allocate(1024);
	private Selector selector;
	private static Logger logger = LoggerFactory.getLogger(DisConfigNIOServer.class);
	private SocketChannel client = null;
	
	public DisConfigNIOServer(int port) {
		this.port = port;
		try {
			init();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void init() throws IOException {
		// 打开ServerSocketChannel，用于监听客户端的连接，它是所有客户端连接的父管道
		ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
		// 设置连接为非阻塞模式
		serverSocketChannel.configureBlocking(false);
		// 绑定监听端口
		ServerSocket serverSocket = serverSocketChannel.socket();
		serverSocket.bind(new InetSocketAddress(port));
		// 创建Reactor线程，创建多路复用器并启动线程
		selector = Selector.open();
		serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
		logger.info("server start on port:" + port);
	}

	@Override
	public void run() {
		while (true) {
			try {
				selector.select();
				Set<SelectionKey> selectionKeys = selector.selectedKeys();
				for (SelectionKey key : selectionKeys) {
					handle(key);
				}
				selectionKeys.clear();
			} catch (Exception e) {
				e.printStackTrace();
				break;
			}

		}
	}
	
	/**
	 * 写入数据 
	 * TODO 这个方法 如果 断线重新连接，不知道是否有问题
	 * @author lfy.xys
	 * @date 2018年5月30日
	 *
	 */
	public void write(byte[] bytes){
		logger.info("write:{}", new String(bytes));
		ByteBuffer outBuffer = ByteBuffer.wrap(bytes);
		try {
			client.write(outBuffer);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	
	private void handle(SelectionKey key) throws IOException, InterruptedException {
		ServerSocketChannel server = null;
		String receiveText = null;
		int count = 0;
		if (key.isAcceptable()) {// 客户端请求连接事件
			logger.info("isAcceptable");
			server = (ServerSocketChannel) key.channel();
			client = server.accept();
			// 判断client 是否为空
			if (client != null) {
				client.configureBlocking(false);
				client.register(selector, SelectionKey.OP_READ);
			}
		} else if (key.isReadable()) {// 获得了可读的事件
			logger.info("isReadable");
			boolean failure = true;

			client = (SocketChannel) key.channel();
			rBuffer.clear();
			try {
				count = client.read(rBuffer);
				failure = false;
			} catch (Exception e) {
				logger.info("断开连接");
				// e.printStackTrace();
			}

			if (count <= 0 || failure) {
				key.cancel(); // Some JDK implementations run into an infinite loop without this.
				logger.info("客户端关闭");
			} else {

				rBuffer.clear();
				byte[] bytes = new byte[count];
				rBuffer.get(bytes, 0, bytes.length);

				client.register(selector, SelectionKey.OP_READ);

				receiveText = new String(bytes);
				logger.info("服务器端收到数据：" + receiveText);

				// 判断是否是 启动标志
				if (SocketUtils.DIS_START_STATE.equals(receiveText)) {
					// TODO 查询到Topic，做成String，以 && 连接
					String topics = "topic_1&&topic_2";

					// 将kafka的topic数组 写回给分发系统
					ByteBuffer outBuffer = ByteBuffer.wrap(topics.getBytes());
					client.write(outBuffer);// 将消息回送给客户端
					logger.info("发送了topics");
				}
			}

		}
	}


	public static void main(String[] args) throws Exception {
		// 创建MyServer
		DisConfigNIOServer server1 = new DisConfigNIOServer(8888);
		new Thread(server1).start();
		
		Thread.sleep(10000);
		server1.write("aaa".getBytes());
	}
}
